package com.ys.http.body;

import com.ys.http.network.SchedulerProvider;
import com.ys.http.upload.DefaultUploadListener;
import com.ys.http.upload.OnUploadListener;

import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import java.io.File;
import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.Disposable;
import okhttp3.MediaType;
import okhttp3.MultipartBody;
import okhttp3.RequestBody;
import okhttp3.internal.Util;
import okio.BufferedSink;
import okio.Okio;
import okio.Source;

public class UploadRequestBody extends RequestBody {

    private MediaType mMediaType = MultipartBody.FORM;
    private OnUploadListener mUploadListener = new DefaultUploadListener();
    private final File mUploadFile;

    private Disposable mProgressDisposable;
    private long mCurrentSize; //当前已读写的字节数
    private long mSecondSize; //上一秒读取的总字节数
    private long mSpeedSize;  //每秒读取的字节数
    private long mTotalSize;

    public static UploadRequestBody create(File file) {
        return new UploadRequestBody(file);
    }

    public static UploadRequestBody create(File file, String mediaType) {
        return new UploadRequestBody(file, mediaType);
    }

    private UploadRequestBody(File file) {
        this.mUploadFile = Objects.requireNonNull(file);
    }

    private UploadRequestBody(File file, String mediaType) {
        this.mUploadFile = Objects.requireNonNull(file);
        if (mediaType != null) {
            mMediaType = MediaType.parse(mediaType);
        }
    }

    private UploadRequestBody(File file, MediaType type) {
        this.mUploadFile = Objects.requireNonNull(file);
        if (type != null) {
            mMediaType = type;
        }
    }

    private void initProgressObserve() {
        mTotalSize = mUploadFile.length();
        mProgressDisposable = Observable.interval(1, TimeUnit.SECONDS)
                .compose(SchedulerProvider.applyScheduler())
                .subscribe(val -> {
                    int progress = 0;
                    if (mTotalSize > 0) {
                        progress = (int) (mCurrentSize * 100 / mTotalSize);
                    }
                    mSpeedSize = mCurrentSize - mSecondSize;
                    mSecondSize += mSpeedSize;
                    mUploadListener.onProgress(progress, mCurrentSize, mSpeedSize);
                });
    }

    public void setOnUploadListener(OnUploadListener listener) {
        if (listener != null) {
            this.mUploadListener = listener;
        }
    }

    @Nullable
    @Override
    public MediaType contentType() {
        return mMediaType;
    }

    @Override
    public void writeTo(@NotNull BufferedSink sink) throws IOException {
        initProgressObserve();
        Source source = Okio.source(mUploadFile);
        try {
            long read;
            while ((read = source.read(sink.getBuffer(), 512*0x400)) != -1) {
                mCurrentSize += read;
                sink.flush();
            }
        } finally {
            Util.closeQuietly(source);
            mProgressDisposable.dispose();
        }
    }

    public void disposable() {
        if (mProgressDisposable != null) {
            mProgressDisposable.dispose();
        }
    }
}
